{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## map(func)\n",
    "\n",
    "Retorna um novo RDD formado pela passagem de cada elemento do RDD de origem através de uma da função `func`.\n",
    "\n",
    "Exemplo:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "data = sc.parallelize(range(1, 11))\n",
    "\n",
    "def pow2(i): return i * i\n",
    "\n",
    "double = data.map(pow2) # ou data.map(lambda i: i * i)\n",
    "\n",
    "print double.collect() "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## filter(func)\n",
    "\n",
    "Retorna um novo RDD formado pela seleção daqueles elemento do RDD de origem que, quando passados para função `func`, retorna `true`.\n",
    "\n",
    "Exemplo:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "data = sc.parallelize(range(1, 11))\n",
    "\n",
    "even = data.filter(lambda i: i % 2 == 0)\n",
    "\n",
    "print even.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## flatMap(func)\n",
    "\n",
    "Semelhante ao map, porém cada item de entrada pode ser mapeado para `0` ou mais itens de saída (assim, func deve retornar uma lista em vez de um único item).\n",
    "\n",
    "Exemplo:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "data = sc.parallelize(range(1, 11))\n",
    "\n",
    "def gt_one_divisors(i): return [j for j in range(2, i) if i % j == 0 ]\n",
    "\n",
    "print 'map:', data.map(gt_one_divisors).collect() \n",
    "\n",
    "print 'flatMap:', data.flatMap(gt_one_divisors).collect() "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## intersection(otherRDD)\n",
    "\n",
    "Retorna um novo RDD que contém a interseção dos elementos no RDD de origem e o outro RDD (argumento).\n",
    "\n",
    "Exemplo:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "two_multiples = sc.parallelize(range(0, 20, 2))\n",
    "\n",
    "three_multiples = sc.parallelize(range(0, 20, 3))\n",
    "\n",
    "print two_multiples.intersection(three_multiples).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## groupByKey()\n",
    "\n",
    "Quando chamado em um RDD de pares `(K, V)`, retorna um conjunto de dados de pares `(K, Iterable<V>)`.\n",
    "\n",
    "Exemplo:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "data = sc.parallelize([ ('a', 1), ('b', 2), ('c', 3) , ('a', 2), ('b', 5), ('a', 3)])\n",
    "\n",
    "for pair in data.groupByKey().collect():\n",
    "    print pair[0], list(pair[1])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## reduceByKey(func)\n",
    "\n",
    "Quando chamado em um RDD de pares `(K, V)`, retorna um RDD de pares `(K, V)` onde os valores de cada chave são agregados usando a função de redução `func`, que deve ser do tipo `(V, V): V` (recebe 2 valores e retorna um novo valor).\n",
    "\n",
    "Exemplo:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "data = sc.parallelize([ ('a', 1), ('b', 2), ('c', 3) , ('a', 2), ('b', 5), ('a', 3)])\n",
    "\n",
    "def summation(a, b): return a + b\n",
    "\n",
    "print data.reduceByKey(summation).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## sortByKey([asceding])\n",
    "\n",
    "Quando chamado em um RDD de pares `(K, V)` em que `K` é ordenável, retorna um RDD de pares `(K, V)` ordenados por chaves em ordem ascendente ou descendente, conforme especificado no argumento `ascending`.\n",
    "\n",
    "Exemplo:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "data = sc.parallelize([ ('a', 1), ('b', 2), ('c', 3) , ('a', 2), ('b', 5), ('a', 3)])\n",
    "\n",
    "def summation(a, b): return a + b\n",
    "\n",
    "print data.sortByKey(ascending=True).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Links:\n",
    "* Lista mais completa com mais funções comuns: http://spark.apache.org/docs/1.6.3/programming-guide.html#transformations\n",
    "* Documentação da API RDD do Spark: http://spark.apache.org/docs/1.6.3/api/python/pyspark.html#pyspark.RDD"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Pyspark (Py 2)",
   "language": "",
   "name": "pyspark"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}
